package org.niit.service

import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.dstream.DStream
import org.niit.bean.AdClickData
import org.niit.dao.{MengRealTimeDao, ZhangRealTimeDao}
//import org.niit.dao.MengRealTimeDao

import org.apache.spark.rdd.OrderedRDDFunctions
import java.io.{FileWriter, PrintWriter}
import java.text.SimpleDateFormat
import java.util.Date
import scala.collection.mutable.ListBuffer

class MengTimeService {

  // 汇总：统一调用的方法
  def dataAnalysis(data: DStream[AdClickData]): Unit = {


    mengTime1(data)

    mengTime2(data)

    mengTime3(data)

    mengTime4(data)
//
    mengTime5(data)
//
    mengTime6(data)


  }




  // 实时分析1：
  def mengTime1(data: DStream[AdClickData]): Unit = {
    // 1. 定义年龄区间划分逻辑
    def getAgeGroup(ageStr: String): String = {
      try {
        val age = ageStr.toInt
        age match {
          case a if a <= 18 => "0-18"
          case a if a <= 35 => "19-35"
          case a if a <= 50 => "36-50"
          case _ => "51+"
        }
      } catch {
        case _: Exception => "未知"
      }
    }

    // 2. 预处理数据：提取时间戳、年龄区间、支付确认页浏览标记
    val processedData = data.map { line =>
      // 时间戳处理：取15秒窗口的整数倍（与15秒窗口对齐）
      val timestamp = line.time.toLong // 假设line.time是毫秒级时间戳
      val windowTs = timestamp / (15 * 1000) * (15 * 1000) // 15秒窗口对齐

      val ageGroup = getAgeGroup(line.age)
      // 标记是否浏览支付确认页（假设"1"表示浏览）
      val isConfirm = if (line.payment_confirmation_page == "1") 1L else 0L

      // (窗口时间戳, 年龄区间) -> (总用户数, 确认页浏览数)
      ((windowTs, ageGroup), (1L, isConfirm))
    }

    // 3. 15秒窗口聚合（每15秒计算一次，窗口大小15秒）
    val windowAgg = processedData.reduceByKeyAndWindow(
      (x: (Long, Long), y: (Long, Long)) => (x._1 + y._1, x._2 + y._2), // 累加总用户数和确认数
      Seconds(15), // 窗口大小：15秒
      Seconds(15)  // 滑动间隔：15秒（每15秒计算一次）
    )

    // 4. 计算转化率并整理格式
    val conversionRate = windowAgg.map { case ((windowTs, ageGroup), (total, confirm)) =>
        val rate = if (total > 0) (confirm.toDouble / total * 100).formatted("%.2f") else "0.00"
        (windowTs, (ageGroup, rate)) // (窗口时间戳, (年龄区间, 转化率))
      }
      // 按时间戳分组，聚合同窗口下的所有年龄区间转化率
      .groupByKey()
      .map { case (ts, ageRateIter) =>
        val ageRateMap = ageRateIter.toMap
        (ts, ageRateMap) // (窗口时间戳, Map(年龄区间 -> 转化率))
      }


        // 5. 输出结果到数据库（修改后：遍历每个年龄区间的转化率，逐个写入）
        conversionRate.foreachRDD { rdd =>
          // 每个分区创建一个数据库连接（避免频繁创建连接，优化性能）
          rdd.foreachPartition { partition =>
            // 在分区内创建数据库访问对象（一次分区创建一次，减少开销）
            val mengRealTimeDao = new MengRealTimeDao
            // 遍历分区中的每条数据
            partition.foreach { case (windowTs, ageRateMap) =>
              // 遍历ageRateMap中的每个键值对（年龄区间 -> 转化率）
              ageRateMap.foreach { case (ageGroup, conversionRate) =>
                // 调用DAO层方法，将单个元素写入数据库
                // 参数说明：窗口时间戳、年龄区间、转化率
                mengRealTimeDao.insertTime1(windowTs.toString, ageGroup, conversionRate)
                // 可选：打印日志确认数据
                println(s"写入数据库 - 窗口时间戳: $windowTs, 年龄区间: $ageGroup, 转化率: $conversionRate%")
              }
            }
          }
        }

    // 5. 输出结果（保持与原有代码一致的文件输出格式）
    conversionRate.foreachRDD { rdd =>
      val list = new ListBuffer[String]
      // 按时间戳排序并收集数据
      val sortedData = rdd.sortByKey(true).collect()

      sortedData.foreach { case (ts, ageRateMap) =>
        // 时间格式化（HH:mm:ss）
        val sdf = new SimpleDateFormat("HH:mm:ss")
        val timeStr = sdf.format(new Date(ts))

        // 组装JSON格式：{ "xtime": "时间", "yval": { "年龄区间": "转化率%", ... } }
        val yvalStr = ageRateMap.map { case (group, rate) =>
          s""""$group": "$rate%""""
        }.mkString("{", ", ", "}")

        list.append(s"""{"xtime": "$timeStr", "yval": $yvalStr}""")

//        // 写入数据库
//        val mengRealTimeDao = new MengRealTimeDao
//        mengRealTimeDao.insertTime1(timeStr,yvalStr)


      }

      // 写入文件（覆盖或追加可根据需求调整）
      val out = new PrintWriter(new FileWriter("output/mengTime1.json"))
      out.println("[" + list.mkString(",") + "]")
      out.flush()
      out.close()
    }
  }


  // 实时分析2：
  def mengTime2(data: DStream[AdClickData]): Unit = {
    // 1. 预处理数据：提取时间窗口、来源渠道、操作系统、主页浏览标记
    val processedData = data.map { line =>
      // 时间戳处理：按15秒窗口对齐（假设line.time为毫秒级时间戳）
      val timestamp = line.time.toLong
      val windowTs = timestamp / (15 * 1000) * (15 * 1000)

      // 提取来源渠道和操作系统
      val source = line.source
      val os = line.operative_system

      // 标记是否浏览主页（假设"1"表示已浏览）
      val isHomeView = if (line.home_page == "1") 1L else 0L

      // 构建聚合键值对：((窗口时间, 来源渠道, 操作系统), (总用户数, 主页浏览数))
      ((windowTs, source, os), (1L, isHomeView))
    }

    // 2. 15秒窗口聚合（与原逻辑保持窗口参数一致）
    val windowAgg = processedData.reduceByKeyAndWindow(
      (x: (Long, Long), y: (Long, Long)) => (x._1 + y._1, x._2 + y._2),
      Seconds(15),
      Seconds(15)
    )

    // 3. 计算转化率并整理格式
    val conversionRate = windowAgg.map { case ((windowTs, source, os), (totalUsers, homeViews)) =>
        val rate = if (totalUsers > 0) {
          (homeViews.toDouble / totalUsers * 100).formatted("%.2f")
        } else {
          "0.00"
        }
        (windowTs, (s"$source-$os", rate)) // 合并渠道和系统为键
      }
      .groupByKey()
      .map { case (ts, iter) =>
        val timeStr = new SimpleDateFormat("HH:mm:ss").format(new Date(ts))
        val rateMap = iter.toMap.map { case (key, rate) => s""""$key": "$rate%"""" }
        (timeStr, rateMap)
      }

    // 4. 保持与原代码相同的输出地址和格式
    conversionRate.foreachRDD { rdd =>
      val list = ListBuffer[String]()
      val sortedData = rdd.sortByKey(true).collect()

      sortedData.foreach { case (time, rateMap) =>
        val yvalStr = rateMap.mkString("{", ", ", "}")
        list.append(s"""{"xtime": "$time", "yval": $yvalStr}""")
        // 写入数据库
        val mengRealTimeDao = new MengRealTimeDao
        mengRealTimeDao.insertTime2(time,yvalStr)
      }

      // 输出到指定文件（可根据需求调整）
      val out = new PrintWriter(new FileWriter("output/mengTime2.json"))
      out.println("[" + list.mkString(",") + "]")
      out.flush()
      out.close()
    }
  }


  // 实时分析3：
  def mengTime3(data: DStream[AdClickData]): Unit = {
    // 1. 预处理数据：提取时间窗口、市场级别、设备类型、列表页浏览标记
    val processedData = data.map { line =>
      // 时间戳处理：按15秒窗口对齐（毫秒级时间戳）
      val timestamp = line.time.toLong
      val windowTs = timestamp / (15 * 1000) * (15 * 1000)

      // 提取市场级别和设备类型
      val market = line.market
      val device = line.device

      // 标记是否浏览列表页（假设"1"表示已浏览）
      val isListingView = if (line.listing_page == "1") 1L else 0L

      // 构建聚合键值对：((窗口时间, 市场级别, 设备类型), (总用户数, 列表页浏览数))
      ((windowTs, market, device), (1L, isListingView))
    }

    // 2. 15秒窗口聚合（与原逻辑保持窗口参数一致）
    val windowAgg = processedData.reduceByKeyAndWindow(
      (x: (Long, Long), y: (Long, Long)) => (x._1 + y._1, x._2 + y._2),
      Seconds(15),
      Seconds(15)
    )

    // 3. 计算比例并整理格式
    val ratioData = windowAgg.map { case ((windowTs, market, device), (totalUsers, listingViews)) =>
        val ratio = if (totalUsers > 0) {
          (listingViews.toDouble / totalUsers * 100).formatted("%.2f")
        } else {
          "0.00"
        }
        (windowTs, (s"$market-$device", ratio)) // 合并市场和设备为键
      }
      .groupByKey()
      .map { case (ts, iter) =>
        val timeStr = new SimpleDateFormat("HH:mm:ss").format(new Date(ts))
        val ratioMap = iter.toMap.map { case (key, ratio) => s""""$key": "$ratio%"""" }
        (timeStr, ratioMap)
      }

    // 4. 保持与原代码相同的输出地址和格式
    ratioData.foreachRDD { rdd =>
      val list = ListBuffer[String]()
      val sortedData = rdd.sortByKey(true).collect()

      sortedData.foreach { case (time, ratioMap) =>
        val yvalStr = ratioMap.mkString("{", ", ", "}")
        list.append(s"""{"xtime": "$time", "yval": $yvalStr}""")

        // 写入数据库
        val mengRealTimeDao = new MengRealTimeDao
        mengRealTimeDao.insertTime3(time,yvalStr)
      }

      // 输出到新文件（避免与原有文件冲突，可根据需求调整）
      val out = new PrintWriter(new FileWriter("output/mengTime3.json"))
      out.println("[" + list.mkString(",") + "]")
      out.flush()
      out.close()
      }
    }



  // 实时分析4：
  def mengTime4(data: DStream[AdClickData]): Unit = {
    // 1. 预处理数据：提取时间窗口、市场级别、新用户标识、支付确认页浏览标记
    val processedData = data.map { line =>
      // 时间戳处理：按15秒窗口对齐（毫秒级时间戳）
      val timestamp = line.time.toLong
      val windowTs = timestamp / (15 * 1000) * (15 * 1000)

      // 提取市场级别和新用户标识（假设"1"表示新用户）
      val market = line.market
      val isNewUser = line.new_user == "1"  // 仅保留新用户数据

      // 标记是否浏览支付确认页（假设"1"表示已浏览）
      val isPaymentView = if (line.payment_confirmation_page == "1") 1L else 0L

      // 构建聚合键值对：仅对新用户进行统计，键为(窗口时间, 市场级别)
      ((windowTs, market), (if (isNewUser) 1L else 0L, if (isNewUser) isPaymentView else 0L))
    }

    // 2. 15秒窗口聚合（与原逻辑保持窗口参数一致）
    val windowAgg = processedData.reduceByKeyAndWindow(
      (x: (Long, Long), y: (Long, Long)) => (x._1 + y._1, x._2 + y._2),  // 累加新用户总数和支付页浏览数
      Seconds(15),
      Seconds(15)
    )

    // 3. 计算比例并整理格式
    val ratioData = windowAgg.map { case ((windowTs, market), (newUserTotal, paymentViews)) =>
        val ratio = if (newUserTotal > 0) {
          (paymentViews.toDouble / newUserTotal * 100).formatted("%.2f")  // 支付页浏览比例 = 浏览数/新用户总数
        } else {
          "0.00"
        }
        (windowTs, (market, ratio))
      }
      .groupByKey()
      .map { case (ts, iter) =>
        val timeStr = new SimpleDateFormat("HH:mm:ss").format(new Date(ts))
        val ratioMap = iter.toMap.map { case (market, ratio) => s""""$market": "$ratio%"""" }
        (timeStr, ratioMap)
      }

    // 4. 保持与原代码相同的输出格式，输出到独立文件
    ratioData.foreachRDD { rdd =>
      val list = ListBuffer[String]()
      val sortedData = rdd.sortByKey(true).collect()

      sortedData.foreach { case (time, ratioMap) =>
        val yvalStr = ratioMap.mkString("{", ", ", "}")
        list.append(s"""{"xtime": "$time", "yval": $yvalStr}""")

        // 写入数据库
        val mengRealTimeDao = new MengRealTimeDao
        mengRealTimeDao.insertTime4(time,yvalStr)


      }

      // 输出到指定文件（区分其他分析结果）
      val out = new PrintWriter(new FileWriter("output/mengTime4.json"))
      out.println("[" + list.mkString(",") + "]")
      out.flush()
      out.close()
    }
  }



  // 实时分析5：
//  例如：10 个用户共浏览了 35 个页面，平均页面数为 35 / 10 = 3.5（小数）
  def mengTime5(data: DStream[AdClickData]): Unit = {
    // 1. 预处理数据：提取时间窗口、设备类型、操作系统、总页面浏览数
    val processedData = data.map { line =>
      // 时间戳处理：按15秒窗口对齐（假设line.time为毫秒级时间戳字符串）
      val timestamp = line.time.toLong
      val windowTs = timestamp / (15 * 1000) * (15 * 1000)

      // 提取设备类型和操作系统
      val device = line.device
      val os = line.operative_system

      // 提取总页面浏览数（转换为整数）
      val totalPages = try {
        line.total_pages_visited.toInt
      } catch {
        case _: Exception => 0
      }

      // 构建聚合键值对：((窗口时间, 设备, 操作系统), (总页面数, 用户数))
      ((windowTs, device, os), (totalPages.toLong, 1L))
    }

    // 2. 15秒窗口聚合（窗口大小15秒，滑动间隔15秒）
    val windowAgg = processedData.reduceByKeyAndWindow(
      (x: (Long, Long), y: (Long, Long)) => (x._1 + y._1, x._2 + y._2),
      Seconds(15),
      Seconds(15)
    )

    // 3. 计算平均页面数并整理格式
    val avgPagesData = windowAgg.map { case ((windowTs, device, os), (totalPages, userCount)) =>
        val avgPages = if (userCount > 0) {
          (totalPages.toDouble / userCount).formatted("%.2f")
        } else {
          "0.00"
        }
        (windowTs, (s"$device-$os", avgPages))
      }
      .groupByKey()
      .map { case (ts, iter) =>
        val timeStr = new SimpleDateFormat("HH:mm:ss").format(new Date(ts))
        val avgPagesMap = iter.toMap.map { case (key, avg) => s""""$key": "$avg"""" }
        (timeStr, avgPagesMap)
      }

    // 4. 保持输出格式与原逻辑一致，输出到指定文件
    avgPagesData.foreachRDD { rdd =>
      val list = ListBuffer[String]()
      val sortedData = rdd.sortByKey(true).collect()

      sortedData.foreach { case (time, avgPagesMap) =>
        val yvalStr = avgPagesMap.mkString("{", ", ", "}")
        list.append(s"""{"xtime": "$time", "yval": $yvalStr}""")
        // 写入数据库
        val mengRealTimeDao = new MengRealTimeDao
        mengRealTimeDao.insertTime5(time,yvalStr)
      }

      val out = new PrintWriter(new FileWriter("output/mengTime5.json"))
      out.println("[" + list.mkString(",") + "]")
      out.flush()
      out.close()
    }
  }





  // 实时分析6：
  def mengTime6(data: DStream[AdClickData]): Unit = {
    // 1. 预处理数据：提取时间窗口、市场级别、列表页浏览标记
    val processedData = data.map { line =>
      // 时间戳处理：按15秒窗口对齐（假设line.time为毫秒级时间戳字符串）
      val timestamp = line.time.toLong
      val windowTs = timestamp / (15 * 1000) * (15 * 1000)

      // 提取市场级别
      val market = line.market

      // 标记是否浏览列表页（"1"表示已浏览）
      val isListingView = if (line.listing_page == "1") 1L else 0L

      // 构建聚合键值对：((窗口时间, 市场级别), 列表页浏览数)
      ((windowTs, market), isListingView)
    }

    // 2. 15秒窗口聚合（窗口大小15秒，滑动间隔15秒）
    val windowAgg = processedData.reduceByKeyAndWindow(
      (x: Long, y: Long) => x + y,  // 累加列表页浏览数
      Seconds(15),
      Seconds(15)
    )

    // 3. 整理格式
    val countData = windowAgg.map { case ((windowTs, market), count) =>
        val timeStr = new SimpleDateFormat("HH:mm:ss").format(new Date(windowTs))
        (timeStr, (market, count.toString))
      }
      .groupByKey()
      .map { case (ts, iter) =>
        val countMap = iter.toMap.map { case (market, count) => s""""$market": "$count"""" }
        (ts, countMap)
      }

    // 4. 保持输出格式与原逻辑一致，输出到指定文件
    countData.foreachRDD { rdd =>
      val list = ListBuffer[String]()
      val sortedData = rdd.sortByKey(true).collect()

      sortedData.foreach { case (time, countMap) =>
        val yvalStr = countMap.mkString("{", ", ", "}")
        list.append(s"""{"xtime": "$time", "yval": $yvalStr}""")

        // 写入数据库
        val mengRealTimeDao = new MengRealTimeDao
        mengRealTimeDao.insertTime6(time,yvalStr)
      }

      val out = new PrintWriter(new FileWriter("output/mengTime6.json"))
      out.println("[" + list.mkString(",") + "]")
      out.flush()
      out.close()
    }
  }





}